package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.StockMinWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.map.StockPutHdfsMap;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.concurrent.TimeUnit;

/**
 * 个股分时数据备份 -> HDFS
 * 1.每分钟生成最新一条数据
 */
public class StockMinHdfsTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 开发步骤：
         * 1.设置HDFS存储路径
         * 2.设置数据文件参数
         *   （大小、分区、格式、前缀、后缀）
         * 3.数据分组
         * 4.划分时间窗口
         * 5.数据处理
         * 6.转换并封装数据
         * 7.写入HDFS
         */
        System.setProperty("HADOOP_USER_NAME","root");

        //1.设置HDFS存储路径
        //设置文件的名称
        OutputFileConfig fileConfig = new OutputFileConfig("stock", ".txt");
        //4.设置文件流对象
        FileSink<String> sink = FileSink
                .forRowFormat(new Path(QuotConfig.config.getProperty("stock.sec.hdfs.path")), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy( //滚动策略，以下三个条件，满足其一就会触发，文件滚动
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(1)) //文件的间隔时间滚动（1分钟滚动生成一个文件）
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(10)) //在一分钟之内，没有往文件发送数据，会触发滚动生成一个新的文件
                                .withMaxPartSize(1024 * 1024 * 1024) //文件的大小，文件大小=1G
                                .build())
                .withOutputFileConfig(fileConfig)
                .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd")) //设置文件的分区策略
                .build();


        //3.数据分组
        waterData.keyBy(new KeyFunction())
        //4.划分时间窗口
        .window(TumblingEventTimeWindows.of(Time.seconds(60)))
        //5.数据处理
        .apply(new StockMinWindowFunction())
        //6.转换并封装数据
        .map(new StockPutHdfsMap())
        //7.写入HDFS
        .sinkTo(sink);


    }
}
